package com.atguigu.flink.chapter10;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
常用 Connector 读写-------file
 */
public class FileConnectorDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //通过ddl的方式创建连接器
        tEnv.executeSql("create table sensor(" +
                "id String," +
                "ts bigint," +
                "vc int" +
                ") with(" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\sensor.json', " +
                " 'format' = 'json' " +
                ")");

        Table result = tEnv.sqlQuery("select * from sensor where id = 'sensor_1'");
        tEnv.executeSql("create table s1(" +
                "a string," +
                "b bigint," +
                "d int" +
                ") with (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\s1.txt', " +
                " 'format' = 'json'" +
                ")");

        result.executeInsert("s1");
    }
}




//public class FileConnectorDemo {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        //通过ddl的方式创建连接器
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with (" +
//                " 'connector' = 'filesystem', " +
//                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\sensor.json', " +
//                " 'format' = 'json' " +
//                ")");
//
//        Table result = tEnv.sqlQuery("select * from sensor where id = 'sensor_2'");
//
//        tEnv.executeSql("create table s1(" +
//                "a string," +
//                "b bigint," +
//                "c int" +
//                ") with(" +
//                " 'connector' = 'filesystem', " +
//                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\s2.txt', " +
//                " 'format' = 'json' " +
//                ")");
//
//        result.executeInsert("s1");
//    }
//}


//public class FileConnectorDemo {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        //通过ddl查询创建连接器
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int" +
//                ") with(" +
//                " 'connector' = 'filesystem', " +
//                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\sensor.json', " +
//                " 'format' = 'json' " +
//                ")");
//
//        Table result = tEnv.sqlQuery("select * from sensor where id = 'sensor_1'");
//
//        tEnv.executeSql("create table s1(" +
//                "a string," +
//                "b bigint," +
//                "c int" +
//                ") with(" +
//                " 'connector' = 'filesystem', " +
//                " 'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\s1.txt', " +
//                " 'format' = 'json' " +
//                ")");
//
//        result.executeInsert("s1");
//    }
//}